package mq;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * @author huanggy
 * @date 2023/02/08
 **/
public class MyKafkaProducer {

    public static final String TOPIC_NAME = "MAIN_DATA_CIGARETTE";

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.50.80.194:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 获取队列
        AdminClient adminClient = AdminClient.create(properties);
        ListTopicsResult result = adminClient.listTopics();
        Set<String> topics = result.names().get();
        System.out.println(topics);

        // 创建队列 - 会自动创建
//        if (!topics.contains(TOPIC_NAME)) {
//            NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, (short) 1);
//            CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
//            System.out.println(createTopicsResult.all().get());
//        }

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        ProducerRecord<String, String> brand = new ProducerRecord<>(TOPIC_NAME, 0,"*",
                "{\n" +
                        "  \"HEAD\" : {\n" +
                        "    \"MSGCODE\" : \"TZ_AC_CGT_BRAND\",\n" +
                        "    \"ELEMENTCODE\" : \"Z_AC_CGT_BRAND\",\n" +
                        "    \"MSGNAME\" : \"卷烟品牌代码\",\n" +
                        "    \"SERVICECODE\" : \"MDM_TO_HYTK\",\n" +
                        "    \"SOURCESYS\" : \"MDM\",\n" +
                        "    \"TARGETSYS\" : \"CG110002\",\n" +
                        "    \"CREATETIME\" : \"2023-02-08 14:48:41\",\n" +
                        "    \"ACTION\" : \"INSERT\",\n" +
                        "    \"MSGTYPE\" : \"1\"\n" +
                        "  },\n" +
                        "  \"DATA\" : [ {\n" +
                        "    \"ACTION\" : \"INSERT\",\n" +
                        "    \"CGT_BRAND_ID\" : \"26e07c6da1d342cdb033e2b2799d134b\",\n" +
                        "    \"CGT_BRAND_CODE\" : \"1289\",\n" +
                        "    \"CGT_BRAND_NAME\" : \"金筒\",\n" +
                        "    \"IS_KEY_BRAND\" : \"\",\n" +
                        "    \"REMARK\" : \"\"\n" +
                        "  } ]\n" +
                        "}");

        producer.send(brand).get();

        ProducerRecord<String, String> spu = new ProducerRecord<>(TOPIC_NAME, 0,"*",
                "{\n" +
                        "  \"HEAD\" : {\n" +
                        "    \"MSGCODE\" : \"TZ_AC_CIGARETTE\",\n" +
                        "    \"ELEMENTCODE\" : \"Z_AC_CIGARETTE\",\n" +
                        "    \"MSGNAME\" : \"卷烟规格基本信息\",\n" +
                        "    \"SERVICECODE\" : \"MDM_TO_HYTK\",\n" +
                        "    \"SOURCESYS\" : \"MDM\",\n" +
                        "    \"TARGETSYS\" : \"CG110002\",\n" +
                        "    \"CREATETIME\" : \"2023-02-08 09:32:36\",\n" +
                        "    \"ACTION\" : \"INSERT\",\n" +
                        "    \"MSGTYPE\" : \"1\"\n" +
                        "  },\n" +
                        "  \"DATA\" : [ {\n" +
                        "    \"ACTION\" : \"INSERT\",\n" +
                        "    \"CGT_CODE\" : \"AC2001841\",\n" +
                        "    \"CGT_NAME\" : \"红双喜(南洋19)\",\n" +
                        "    \"CGT_MIN_CODE\" : \"4891132580012\",\n" +
                        "    \"CGT_BRAND_CODE\" : \"1289\",\n" +
                        "    \"CGT_ORG_CODE\" : \"18000001\",\n" +
                        "    \"CGT_TYPE_CODE\" : \"02\",\n" +
                        "    \"CGT_PACKAGE_DIFFENT\" : \"\",\n" +
                        "    \"IS_LOW_PRICE\" : \"N\",\n" +
                        "    \"CGT_TOTALLENGTH\" : \"97\",\n" +
                        "    \"CGT_LENGTH\" : \"67\",\n" +
                        "    \"CGT_FILTERLENGTH\" : \"30\",\n" +
                        "    \"CGT_GIRTH\" : \"17\",\n" +
                        "    \"CGT_TARCONTENT\" : \"7\",\n" +
                        "    \"CGT_NICOTINIC\" : \"1\",\n" +
                        "    \"CGT_CO\" : \"\",\n" +
                        "    \"CGT_IMPORTFLAG_CODE\" : \"1\",\n" +
                        "    \"CGT_FILLTERTYPE_CODE\" : \"1\",\n" +
                        "    \"CGT_IS_PEARL\" : \"1\",\n" +
                        "    \"CGT_IS_TINY\" : \"1\",\n" +
                        "    \"CGT_IS_MEDIUM_BRANCE\" : \"1\",\n" +
                        "    \"CGT_IS_SGORT_BRANCE\" : \"0\",\n" +
                        "    \"CGT_NO_TAX_ALLOT_PRICE\" : \"0\",\n" +
                        "    \"CGT_APRICE\" : \"0\",\n" +
                        "    \"CGT_WPRICE\" : \"0\",\n" +
                        "    \"CGT_RPRICE\" : \"0\",\n" +
                        "    \"CGT_PR_TYPE_CODE\" : \"302\",\n" +
                        "    \"CGT_PR_LEVEL_CODE\" : \"\",\n" +
                        "    \"CGT_RECORDDATE\" : \"2019-02-12\",\n" +
                        "    \"CGT_INSTRUCTION\" : \"\",\n" +
                        "\"SUB_TABLE_LIST\" : {\n" +
                        "      \"TZ_AC_CIGARETTE_SPECS_PACK\" : [ {\n" +
                        "        \"CGT_PACK_CODE\" : \"AC3200511\",\n" +
                        "        \"CGT_CODE\" : \"AC2001841\",\n" +
                        "        \"CGT_BARCARRIER_CODE\" : \"件\",\n" +
                        "        \"CGT_MAINCOLOR\" : \"玫瑰金\",\n" +
                        "        \"CGT_SECONDCOLOR\" : \"金\",\n" +
                        "        \"CGT_PACKTYPE_CODE\" : \"1003\",\n" +
                        "        \"CGT_PACKAMOUNT\" : \"10000\",\n" +
                        "        \"CGT_NEWCODEBAR\" : \"\",\n" +
                        "        \"CGT_NEWINNERBARCODE\" : \"\",\n" +
                        "        \"CGT_INNERBARCODE\" : \"\",\n" +
                        "        \"CGT_CODE_OLD\" : \"4891132580036\"\n" +
                        "      } ]\n" +
                        "    }\n" +
                        "  } ]\n" +
                        "}");

        producer.send(spu).get();
//        System.out.println("同步方式得到的消息：" + "topic-" + recordMetadata.topic()
//                + "|partition-" + recordMetadata.partition()
//                + "|offset-" + recordMetadata.offset());

    }
}
